package com.cetc.sdp.kmga.cs.stream;

import com.alibaba.fastjson.JSON;
import com.cetc.sdp.kmga.cs.audit.AlertReason;
import com.cetc.sdp.kmga.cs.audit.AuditRuleFastMatcher;
import com.cetc.sdp.kmga.cs.common.IndexTable;
import com.cetc.sdp.kmga.cs.common.RabbitMQConf;
import com.cetc.sdp.kmga.cs.common.TableMapping;
import com.cetc.sdp.kmga.cs.common.TableMeta;
import com.cetc.sdp.kmga.cs.jdbc.AuditDAO;
import com.cetc.sdp.kmga.cs.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;

import java.io.IOException;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;

/**
 * @description:
 * @author： DengQiang
 * @date: 2018/3/23 11:26
 */
public class StreamUtils {

    /**
     * 补齐id
     *
     * @param sb
     * @param size
     * @param suffix
     */
    public static void idAppend(StringBuilder sb, int size, String suffix) {
        for (; sb.length() < size; ) {
            sb.append(suffix);
        }
    }

    private static Random random = new Random(System.currentTimeMillis());

    /**
     * 产生1-9位的随机数
     *
     * @param len 长度
     * @return
     */
    public static String random(int len) {
        switch (len) {
            case 1:
                return String.format("%d", random.nextInt(10));
            case 2:
                return String.format("%02d", random.nextInt(100));
            case 3:
                return String.format("%03d", random.nextInt(1000));
            case 4:
                return String.format("%04d", random.nextInt(10000));
            case 5:
                return String.format("%05d", random.nextInt(100000));
            case 6:
                return String.format("%06d", random.nextInt(1000000));
            case 7:
                return String.format("%07d", random.nextInt(10000000));
            case 8:
                return String.format("%08d", random.nextInt(100000000));
            case 9:
                return String.format("%09d", random.nextInt(1000000000));
            default:
                return "0";
        }
    }

    /**
     * 重置累加器
     *
     * @param accumulator
     * @param initialSeqValue
     */
    public static void resetAccumulator(LongAccumulator accumulator, long initialSeqValue) {
        accumulator.reset();
        accumulator.add(initialSeqValue);
    }

    /**
     * 发送告警信息
     *
     * @param reasons
     * @param conf
     */
    public static void sendAlert(List<AlertReason> reasons, RabbitMQConf conf) {
        ThreadPool.getExecutorService().submit(() -> {
            RabbitMQProducer producer = RabbitMQProducer.getInstance(conf.getNodes(), conf.getMqUsername(), conf.getMqPassword(), conf.getMqAuditAlertExch(), conf.getMqAuditAlertQueue(), RabbitMQProducer.XT.DEFAULT);
            reasons.forEach(reason -> producer.sendMsg(JSON.toJSONString(reason)));
        });
    }

    /**
     * 初始化id递增序号
     *
     * @param tm
     * @param date
     * @param changeByMonth
     * @return
     */
    public static long getInitialSeqForTable(TableMapping tm, LocalDate date, boolean changeByMonth) {
        if (tm.getInitial() > tm.getCurrent()) {
            return tm.getInitial();
        } else {
            if (changeByMonth && (tm.getLastUpdateTime().getMonthValue() != date.getMonthValue() ||
                    tm.getLastUpdateTime().getYear() != date.getYear())) {
                return tm.getInitial();
            } else {
                return tm.getCurrent();
            }
        }
    }

    public static long getInitialSeqForTable(TableMeta tm, LocalDate date, boolean changeByMonth) {
        if (tm.getInitial() > tm.getCurrent()) {
            return tm.getInitial();
        } else {
            if (changeByMonth && (tm.getLastUpdateTime().getMonthValue() != date.getMonthValue() ||
                    tm.getLastUpdateTime().getYear() != date.getYear())) {
                return tm.getInitial();
            } else {
                return tm.getCurrent();
            }
        }
    }

    private static Map<String, Broadcast<IndexTable>> indexTableMap = new ConcurrentHashMap<>();

    /**
     * 审计规则广播变量包装类
     */
    public static class RuleFastMatcherBroadcastWrapper extends AbstractBroadcastWrapper<AuditRuleFastMatcher> implements MessageListener {

        private static volatile RuleFastMatcherBroadcastWrapper INSTANCE;

        private RuleFastMatcherBroadcastWrapper(JavaSparkContext jsc) {
            super(jsc);
        }

        private static final String AUDIT_MSG_TYPE_PREFIX = "Audit";

        public static RuleFastMatcherBroadcastWrapper getInstance(JavaSparkContext jsc) {
            if (INSTANCE == null) {
                synchronized (RuleFastMatcherBroadcastWrapper.class) {
                    if (INSTANCE == null) {
                        INSTANCE = new RuleFastMatcherBroadcastWrapper(jsc);
                    }
                }
            }
            return INSTANCE;
        }

        @Override
        public void fireMessage(String message) {
            System.out.println("Recv :" + message);
            try {
                Map<String, Object> msgObj = (Map<String, Object>) JSON.parse(message);
                String type = String.valueOf(msgObj.get("type"));
                if (null != type && type.startsWith(AUDIT_MSG_TYPE_PREFIX)) {
                    refreshBroadcast();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        protected Broadcast<AuditRuleFastMatcher> broadcast(JavaSparkContext jsc) {
            AuditRuleFastMatcher matcher = new AuditRuleFastMatcher(AuditDAO.findAllEnabledRules(), AuditDAO.findAllResource(),
                    AuditDAO.findAllOpCode(), AuditDAO.findAllRiskMap());
            return jsc.broadcast(matcher);
        }
    }

    /**
     * 广播变量Wrapper
     * 客户端采集集群查找表，key客户端ip，值为Tuple，其中第0个值是集群id, 第1值是集群类型
     */
    public static class ClientIpClusterBroadcastWrapper extends AbstractBroadcastWrapper<Map<String, Tuple2<Integer, Integer>>> implements MessageListener {

        private static volatile ClientIpClusterBroadcastWrapper INSTANCE;

        private ClientIpClusterBroadcastWrapper(JavaSparkContext jsc) {
            super(jsc);
        }

        private static final String CLIENT_IP_CLUSTER_MSG_TYPE_PREFIX = "ClientIpCluster";

        public static ClientIpClusterBroadcastWrapper getInstance(JavaSparkContext jsc) {
            if (INSTANCE == null) {
                synchronized (ClientIpClusterBroadcastWrapper.class) {
                    if (INSTANCE == null) {
                        INSTANCE = new ClientIpClusterBroadcastWrapper(jsc);
                    }
                }
            }
            return INSTANCE;
        }

        @Override
        protected Broadcast<Map<String, Tuple2<Integer, Integer>>> broadcast(JavaSparkContext jsc) {
            Map<String, int[]> map = AuditDAO.findAllClientCluster();
            Map<String, Tuple2<Integer, Integer>> bmap = new HashMap<>(map.size());
            map.forEach((key, value) -> {
                bmap.put(key, new Tuple2<>(value[0], value[1]));
            });
            return jsc.broadcast(bmap);
        }

        @Override
        public void fireMessage(String message) {
            System.out.println("Recv :" + message);
            try {
                Map<String, Object> msgObj = (Map<String, Object>) JSON.parse(message);
                String type = String.valueOf(msgObj.get("type"));
                if (null != type && type.startsWith(CLIENT_IP_CLUSTER_MSG_TYPE_PREFIX)) {
                    refreshBroadcast();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * id 递增序列生成类
     */
    public static class JavaIdGenerator {

        private static volatile Map<String, LongAccumulator> accumulatorMap = new ConcurrentHashMap<>();

        public static LongAccumulator getInstance(JavaSparkContext jsc, String name, long initial) {
            LongAccumulator instance = accumulatorMap.get(name);
            if (instance == null) {
                synchronized (JavaIdGenerator.class) {
                    instance = accumulatorMap.get(name);
                    if (instance == null) {
                        instance = jsc.sc().longAccumulator(name);
                        instance.add(initial);
                        accumulatorMap.put(name, instance);
                    }
                }
            }
            return instance;
        }
    }

    /**
     * 配置HBase Job
     *
     * @param jsc
     * @param prop
     * @return
     */
    public static Job configJob(JavaSparkContext jsc, Map<String, String> prop) {
        Job hbaseJob;
        try {
            hbaseJob = Job.getInstance(jsc.hadoopConfiguration());
            Configuration conf = hbaseJob.getConfiguration();
            prop.forEach((k, v) -> conf.set(k.toString(), v.toString()));
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
        hbaseJob.setOutputKeyClass(ImmutableBytesWritable.class);
        hbaseJob.setOutputValueClass(Result.class);
        hbaseJob.setOutputFormatClass(PooledMultiTableOutputFormat.class);
        return hbaseJob;
    }

    /**
     * 获取id逆序时间戳
     *
     * @param logTime
     * @return
     */
    public static long getIdTimestamp(LocalDateTime logTime) {
        if (logTime != null) {
            return Long.MAX_VALUE - Tool.dateToTimestamp(logTime);
        } else {
            return Long.MAX_VALUE - Instant.now().atZone(Tool.zoneId).toInstant().toEpochMilli();
        }
    }

    private static final Pattern COMMA = Pattern.compile(",");

    public static RabbitMQConf createRabbitMQConf(Map<String, String> rabbitmqConf) {
        String mqUsername = rabbitmqConf.get("rabbitmq.username");
        String mqPassword = rabbitmqConf.get("rabbitmq.password");
        String mqNotifyQueue = rabbitmqConf.get("rabbitmq.notify.queue");
        String mqNotifyExchange = rabbitmqConf.get("rabbitmq.notify.exchange");
        String mqDevExchange = rabbitmqConf.get("rabbitmq.dev.exchange");
        String mqAuditAlertExch = rabbitmqConf.get("rabbitmq.audit.alert.exchange");
        String mqAuditAlertQueue = rabbitmqConf.get("rabbitmq.audit.alert.queue");
        String mqLogQueue = rabbitmqConf.get("rabbitmq.log.queue");
        String mqNodeList = rabbitmqConf.get("rabbitmq.node.list");
        RabbitMQConf mqConf = new RabbitMQConf();
        mqConf.setMqUsername(mqUsername);
        mqConf.setMqPassword(mqPassword);
        mqConf.setMqNotifyQueue(mqNotifyQueue);
        mqConf.setMqNotifyExchange(mqNotifyExchange);
        mqConf.setMqDevExchange(mqDevExchange);
        mqConf.setMqAuditAlertQueue(mqAuditAlertQueue);
        mqConf.setMqAuditAlertExch(mqAuditAlertExch);
        String[] nodes = COMMA.split(mqNodeList);
        for (String node : nodes) {
            String[] ip = node.split(":");
            if (ip.length == 2) {
                mqConf.addNode(ip[0], Integer.parseInt(ip[1]));
            }
        }
        String[] queue = COMMA.split(mqLogQueue);
        for (String q : queue) {
            mqConf.addLogQueue(q);
        }
        return mqConf;
    }

    public static long getIdTimestamp(String logTime) {
        return getIdTimestamp(Tool.getTimeFromOrDefault(logTime));
    }

    /**
     * 为了让apache phoenix 查询habse数据，必须添加以下列
     */
    public final static byte[] PHOENIX_APPEND_COL = Bytes.toBytes("_0");
    public final static byte[] PHOENIX_APPEND_COL_VALUE = Bytes.toBytes("x");

    public static final byte[] INDEX_COLUMN_FAMILY_BYTES = Bytes.toBytes("C1");
    public static final byte[] INDEX_COLUMN_ROWKEY_BYTES = Bytes.toBytes("ROWKEY");
}
